package com.xuliugen.storm.demo.trident.operator.filter;

import com.xuliugen.storm.demo.trident.model.DiagnosisEvent;
import org.apache.storm.trident.operation.BaseFilter;
import org.apache.storm.trident.tuple.TridentTuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 实现BaseFilter的方法就可以对tuple进行过滤
 */
public class DiseaseFilter extends BaseFilter {

    private static final long serialVersionUID = 1L;
    private static final Logger logger = LoggerFactory.getLogger(DiseaseFilter.class);

    /**
     * Filter操作结果返回True的tuple将会被发送到下游进行操作；
     * 返回False该tuple就不会发送到下游
     * @param tuple
     * @return
     */
    @Override
    public boolean isKeep(TridentTuple tuple) {
        DiagnosisEvent diagnosis = (DiagnosisEvent) tuple.getValue(0);
        Integer code = Integer.parseInt(diagnosis.diagnosisCode);
        if (code.intValue() <= 322) {
            logger.debug("Emitt ing disease [" + diagnosis.diagnosisCode + "]");
            return true;
        } else {
            logger.debug("Filtering disease [" + diagnosis.diagnosisCode + "]");
            return false;
        }
    }
}